package com.atguigu.flink0624.chapter11.function;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/11/20 10:29
 */
public class Flink02_Function_Table {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        Table table = tenv.fromValues("hello world", "hello atguigu", "hello", "hello hello atguigu");
        tenv.createTemporaryView("t1", table);
        
        // 1. 在table api中使用
        // 1.1. 内联使用
       /* table
            .joinLateral(call(MySplit.class, $("f0")))
            //            .leftOuterJoinLateral(call(MySplit.class, $("f0")))
            .select($("f0"), $("word"), $("len"))
            .execute()
            .print();*/
        
        // 1.2. 注册后使用
        
        // 2. 在sql语句中使用
        tenv.createTemporaryFunction("my_split", MySplit.class);
        /*tenv
            .sqlQuery("select" +
                          " f0, " +
                          " w, " +
                          " l " +
                          "from t1 " +
                          "join lateral table(my_split(f0)) as T(w, l) on true")
            .execute()
            .print();*/
        // select ... from a join b on a.id=b.id  内连接
        // select ... from a, b where a.id=b.id  内连接
    
        /*tenv
            .sqlQuery("select" +
                          " f0, " +
                          " w, " +
                          " l " +
                          "from t1, " +
                          "lateral table(my_split(f0)) as T(w, l)")
            .execute()
            .print();*/
    
        tenv
            .sqlQuery("select" +
                          " f0, " +
                          " w, " +
                          " l " +
                          "from t1  " +
                          "left join lateral table(my_split(f0)) as T(w, l) on true")
            .execute()
            .print();
        
    }
    
    @FunctionHint(output = @DataTypeHint("row<word string, len int>"))
    public static class MySplit extends TableFunction<Row> {
        public void eval(String s) {
            if (s.length() < 8) {
                return;
            }
            
            String[] words = s.split(" ");
            
            for (String word : words) {
                collect(Row.of(word, word.length()));
            }
        }
    }
    
}
/*
hello world      hello    5
                 world    5

hello atguigu    ...


 */
